package com.orangeandbronze.demos.mapreduce;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import com.google.appengine.tools.mapreduce.Reducer;
import com.google.appengine.tools.mapreduce.ReducerInput;

public class PhrasesReducer extends Reducer<String, String, ByteBuffer> {

	private static final long serialVersionUID = -3822878848346467538L;

	@Override
	public void reduce(String key, ReducerInput<String> values) {
		int length = 0;
		List<String> filenames = new ArrayList<String>();
		while (values.hasNext()) {
			filenames.add(values.next());
			length++;
		}
		if (length < 10) {
			return; // do nothing
		}
		Map<String, Long> counts = new HashMap<String, Long>();
		for (String filename : filenames) {
			counts.put(filename,
					(counts.containsKey(filename) ? counts.get(filename) : 0) + 1);
		}
		String words = key.replaceAll(":", " ");
		int threshold = length / 2;
		for (Entry<String, Long> entry : counts.entrySet()) {
			if (entry.getValue() > threshold) {
				getContext().emit(ByteBuffer.wrap(
						(words + ": " + entry.getKey() + "\n").getBytes()));
			}
		}
	}

	/*
def phrases_reduce(key, values):
  """Phrases demo reduce function."""
  if len(values) < 10:
    return
  counts = {}
  for filename in values:
    counts[filename] = counts.get(filename, 0) + 1

  words = re.sub(r":", " ", key)
  threshold = len(values) / 2
  for filename, count in counts.items():
    if count > threshold:
      yield "%s:%s\n" % (words, filename)

 	 */
}
